package com.hucais.etl.common.service

import java.{lang, util}

import cn.hutool.core.date.DateUtil
import com.hucais.core.constant.Constants
import com.hucais.core.utils.CommonUtils
import com.hucais.etl.common.bean.{DDCategory, DdBookSelectInfo, DwdBookBaseInfo, DwdBookBaseInfoWithAuthorPopularity}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{Dataset, SparkSession}

import scala.math.BigDecimal.RoundingMode

/**
 * 选品模型-通用服务
 */
object CommonSelectService {

  /**
   * 计算选品模型积分
   *
   * @param sparkSession
   * @param bookDS               图书基本数据
   * @param publishingHouseMapBd 出版社选品映射
   * @param categoryWeightMapBd  各分类权重映射
   * @param universityMapBd      大学高校数据Map
   * @return
   */
  def calSelectInfoIntegral(sparkSession: SparkSession, bookDS: Dataset[DwdBookBaseInfoWithAuthorPopularity],
                            publishingHouseMapBd: Broadcast[util.Map[String, lang.Long]],
                            categoryWeightMapBd: Broadcast[util.Map[String, DDCategory]],
                            universityMapBd: Broadcast[util.Map[String, util.List[String]]]): Dataset[DdBookSelectInfo] = {
    import sparkSession.implicits._
    bookDS.mapPartitions(partitions => {
      partitions.map(item => {
        // 获取各维度的权重
        val categoryWeightMap = categoryWeightMapBd.value
        val categoryWeight = categoryWeightMap.get(item.first_category + "_" + item.second_category)
        val yearSalesWeight = categoryWeight.getYear_sales_weight
        val authorPopularityWeight = categoryWeight.getAuthor_popularity_weight
        val doubanScoreWeight = categoryWeight.getDouban_score_weight
        val publishingHouseWeight = categoryWeight.getPublishing_house_weight
        val brandPlus = categoryWeight.getBrand_plus

        //计算各维度积分
        val year_sales_integral = CommonSelectService.calYearSalesIntegral(item.year_sales, yearSalesWeight)
        val doubanScoreIntegral = CommonSelectService.calDoubanScoreIntegral(item.grader_cnt, item.douban_score, doubanScoreWeight)
        val publishingHouseIntegral = CommonSelectService.calPublishingHouseIntegral(publishingHouseMapBd.value, item.publishing_house, publishingHouseWeight)
        val brandIntegral = CommonSelectService.calBrandIntegral(item.brand, brandPlus)

        var authorPopularityIntegral = 0.00F
        if (StringUtils.isNotBlank(item.author)) {
          authorPopularityIntegral = CommonSelectService.calAuthorPopularityIntegral(item, authorPopularityWeight, universityMapBd.value)
        }

        //计算总积分
        val total_integral = (
          BigDecimal.decimal(year_sales_integral) + BigDecimal.decimal(doubanScoreIntegral) + BigDecimal.decimal(publishingHouseIntegral) +
            BigDecimal.decimal(authorPopularityIntegral) + BigDecimal.decimal(brandIntegral)
          ).setScale(2, RoundingMode.HALF_UP).floatValue()

        //组装结果数据
        val currentTime = DateUtil.format(new java.util.Date(), "yyyy-MM-dd HH:mm:ssZ").replace(" ", "T")
        DdBookSelectInfo(
          Constants.FIRST_CHANNEL_REVERSE_SELECT, Constants.SECOND_CHANNEL_DD, item.isbn, item.book_name, item.first_category,
          item.second_category, item.third_category, item.fourth_category, item.author, item.publishing_house,
          item.publishing_time, item.selling_price, item.min_selling_price,item.avg_selling_price,item.store_cnt,
          item.store_pricing,item.year_sales, item.discount,item.brand, item.douban_score.toFloat, year_sales_integral,
          authorPopularityIntegral, doubanScoreIntegral,publishingHouseIntegral, brandIntegral, total_integral, currentTime
        )
      })
    })
  }

  /**
   * 计算品牌积分
   *
   * @param brand     品牌
   * @param brandPlus 品牌得分
   * @return
   */
  def calBrandIntegral(brand: String, brandPlus: lang.Double): Float = {
    if (CommonUtils.isNotBlankExt(brand) && !brand.equals("无")) brandPlus.floatValue() else 0.00F
  }

  /**
   * 计算作者知名度积分
   *
   * @param bookInfo               图书详情
   * @param authorPopularityWeight 作者知名度的权重
   * @param universityMap          大学高校数据
   * @return
   */
  def calAuthorPopularityIntegral(bookInfo: DwdBookBaseInfoWithAuthorPopularity, authorPopularityWeight: Double,
                                  universityMap: util.Map[String, util.List[String]]): Float = {
    var author_popularity_integral: BigDecimal = 0.00

    val publishedBookWeight = 0.20
    val fanWeight = 0.50
    val universityWeight = 0.30
    val followers_unit = 10000

    //计算历史出书-积分
    var publishedBookIntegral = 0
    val published_num = bookInfo.published_num
    if (published_num > 0) {
      published_num match {
        case published_num if published_num >= 10 => publishedBookIntegral = 10
        case published_num if 5 until 10 contains published_num => publishedBookIntegral = 8
        case published_num if 3 until 5 contains published_num => publishedBookIntegral = 5
        case published_num if published_num < 3 => publishedBookIntegral = 3
        case _ => publishedBookIntegral = 0
      }
    }

    //计算粉丝量-积分
    var fanIntegral = 0
    val total_followers = bookInfo.total_followers
    if (total_followers > 0) {
      total_followers match {
        case total_followers if total_followers > (100 * followers_unit) => fanIntegral = 10
        case total_followers if (50 * followers_unit) until (100 * followers_unit) contains total_followers => fanIntegral = 8
        case total_followers if (10 * followers_unit) until (50 * followers_unit) contains total_followers => fanIntegral = 5
        case total_followers if (2 * followers_unit) until (10 * followers_unit) contains total_followers => fanIntegral = 3
        case total_followers if total_followers < (2 * followers_unit) => fanIntegral = 2
        case _ => fanIntegral = 0
      }
    }

    //计算毕业院校-积分
    var universityIntegral = 0
    if (StringUtils.isNotBlank(bookInfo.university)) {
      universityIntegral = CommonSelectService.calUniversityIntegral(bookInfo.university, universityMap)
    }

    // 计算作者知名度总积分
    author_popularity_integral = (publishedBookIntegral * publishedBookWeight) + (fanIntegral * fanWeight) + (universityIntegral * universityWeight)
    (author_popularity_integral * authorPopularityWeight).setScale(2, RoundingMode.HALF_UP).floatValue()
  }

  /**
   * 计算毕业院校积分
   *
   * @param university    作者的毕业院校
   * @param universityMap 大学高校名单
   * @return
   */
  private def calUniversityIntegral(university: String, universityMap: util.Map[String, util.List[String]]): Int = {
    val universityIntegral = 5
    val type_985 = "985"
    val type_211 = "211"

    val type_985Map = universityMap.get(type_985)
    val type_211Map = universityMap.get(type_211)
    val universityArr = university.split(",")
    for (u <- universityArr) {
      if (type_985Map.contains(u)) {
        return 10
      }
      if (type_211Map.contains(u)) {
        return 8
      }
    }

    universityIntegral
  }

  /**
   * 计算出版社积分
   *
   * @param publishingHouseMap    出版社销量映射
   * @param publishing_house      出版社名称
   * @param publishingHouseWeight 出版社的权重
   * @return
   */
  def calPublishingHouseIntegral(publishingHouseMap: util.Map[String, java.lang.Long], publishing_house: String,
                                 publishingHouseWeight: Double): Float = {
    var publishing_house_integral: BigDecimal = 0.00
    // 统计单位:万
    val unit_wan = 10000

    if (publishingHouseMap.containsKey(publishing_house)) {
      val sales = publishingHouseMap.get(publishing_house)
      sales match {
        case sales if sales > 30 * unit_wan => publishing_house_integral = 10.00
        case sales if 20 * unit_wan until 30 * unit_wan contains sales => publishing_house_integral = 9.00
        case sales if 15 * unit_wan until 20 * unit_wan contains sales => publishing_house_integral = 7.00
        case sales if 10 * unit_wan until 15 * unit_wan contains sales => publishing_house_integral = 5.00
        case sales if 2 * unit_wan until 10 * unit_wan contains sales => publishing_house_integral = 3.00
        case sales if sales < 2 * unit_wan => publishing_house_integral = 2.00
        case _ => publishing_house_integral = 0
      }
    }
    (publishing_house_integral * publishingHouseWeight).setScale(2, RoundingMode.HALF_UP).floatValue()
  }


  /**
   * 计算豆瓣评分积分
   *
   * @param grader_cnt        评价人数
   * @param doubanScore       豆瓣评分
   * @param doubanScoreWeight 豆瓣评分的权重
   * @return
   */
  def calDoubanScoreIntegral(grader_cnt: Long, doubanScore: Double, doubanScoreWeight: Double): Float = {
    var douban_score_integral: BigDecimal = 0.00

    doubanScore match {
      case grader_cnt if (grader_cnt < 100) => douban_score_integral = 0.00
      case grader_cnt if (grader_cnt >= 100) => douban_score_integral = BigDecimal.decimal(doubanScore)
      case _ => douban_score_integral = 0.00
    }
    (douban_score_integral * doubanScoreWeight).setScale(2, RoundingMode.HALF_UP).floatValue()
  }

  /**
   * 计算年销量积分
   *
   * @param yearSales       年销量
   * @param yearSalesWeight 年销量的权重
   * @return
   */
  def calYearSalesIntegral(yearSales: Long, yearSalesWeight: Double): Float = {
    var year_sales_integral: BigDecimal = 0
    yearSales match {
      case yearSales if (yearSales >= 5000) => year_sales_integral = 10
      case yearSales if 2000 until 5000 contains yearSales => year_sales_integral = 8
      case yearSales if 500 until 2000 contains yearSales => year_sales_integral = 5
      case yearSales if 100 until 500 contains yearSales => year_sales_integral = 3
      case yearSales if (yearSales < 100) => year_sales_integral = 2
      case _ => year_sales_integral = 0
    }
    (year_sales_integral * yearSalesWeight).setScale(2, RoundingMode.HALF_UP).floatValue()
  }

  /**
   * 补充豆瓣评分数据
   *
   * @param sparkSession
   * @return
   */
  def fixDoubanScore(sparkSession: SparkSession): Dataset[DwdBookBaseInfo] = {
    import sparkSession.implicits._
    sparkSession.sql(
      s"""
         |select
         |	a.first_channel,a.second_channel,a.isbn,a.book_name,a.first_category,
         |  a.second_category,a.third_category,a.fourth_category,a.author,
         |	a.publishing_house,a.publishing_time,a.selling_price,a.min_selling_price,a.avg_selling_price,
         |	a.store_cnt,a.store_pricing,a.year_sales,a.discount,a.brand,
         |  coalesce(b.score,0.00) as douban_score,coalesce(b.grader_cnt,0) as grader_cnt,a.create_time
         |from book a left join douban b on a.book_name=b.query_book
         |""".stripMargin).as[DwdBookBaseInfo]
  }
}
